package com.atguigu.day08;

import com.atguigu.day03.Example8;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoProcessFunction;
import org.apache.flink.util.Collector;

/**
 * 开关流
 */
public class Example2 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStreamSource<Example8.SensorReading> sensorStream = env.addSource(new Example8.SensorSource());
        DataStreamSource<Tuple2<String, Long>> swichStream = env.fromElements(Tuple2.of("sensor_0", 10 * 1000L));
        swichStream
                .keyBy(x -> x.f0)
                .connect(
                        sensorStream.keyBy(x -> x.sensorId)
                )
                .process(new CoProcessFunction<Tuple2<String, Long>, Example8.SensorReading, Example8.SensorReading>() {
                    private ValueState<Boolean> swichFlag;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        super.open(parameters);
                        swichFlag = getRuntimeContext().getState(new ValueStateDescriptor<Boolean>("swich", Types.BOOLEAN));
                    }

                    //开关流
                    @Override
                    public void processElement1(Tuple2<String, Long> value, Context ctx, Collector<Example8.SensorReading> out) throws Exception {
                        swichFlag.update(true);
                        ctx.timerService().registerProcessingTimeTimer(ctx.timerService().currentProcessingTime() + value.f1);
                    }

                    @Override
                    public void processElement2(Example8.SensorReading value, Context ctx, Collector<Example8.SensorReading> out) throws Exception {
                        if (swichFlag.value() != null && swichFlag.value()) out.collect(value);
                    }

                    @Override
                    public void onTimer(long timestamp, OnTimerContext ctx, Collector<Example8.SensorReading> out) throws Exception {
                        super.onTimer(timestamp, ctx, out);
                        swichFlag.clear();
                    }
                })
                .print();

        env.execute();

    }
}
